package net.bwie.realtime.dwd.douyin.logs.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dwd.douyin.logs.function.IsNewCommentFunction;
import net.bwie.realtime.dwd.douyin.logs.function.IsNewFanFunction;
import net.bwie.realtime.dwd.douyin.logs.function.IsNewUserOrderFunction;
import net.bwie.realtime.dwd.douyin.logs.function.IsNotFanFunction;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class NewFanTag {
    public static void main(String[] args) throws Exception{
        //1-执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        //2-数据源
        DataStream<String> eventLogStream = KafkaUtil.consumerKafka(env, "dwd_event_log");
        DataStream<String> tradeOrderStream = KafkaUtil.consumerKafka(env, "dwd_trade_order");
        DataStream<String> fanRelationStream = KafkaUtil.consumerKafka(env, "dwd_fan_relation");

        DataStream<String> processLogStream = processLog(eventLogStream, tradeOrderStream, fanRelationStream);
        processLogStream.print();

        env.execute("NewFanTag");
    }

    private static DataStream<String> processLog(DataStream<String> eventLogStream,
                                              DataStream<String> tradeOrderStream,
                                              DataStream<String> fanRelationStream) {
        //1.用户是否第一次下单
        DataStream<String> isNewStream = isNewUserOrder(tradeOrderStream);
        KafkaUtil.producerKafka(isNewStream,"dwd_user_first_detail");
        //2.是否新增评论
        DataStream<String> isNewCommentStream = isNewComment(eventLogStream);
        KafkaUtil.producerKafka(isNewCommentStream,"dwd_user_new_comment_detail");
        //3.是否新增粉丝
        DataStream<String> isNewFanStream = isNewFan(fanRelationStream);
        KafkaUtil.producerKafka(isNewFanStream,"dwd_user_new_fans_detail");
        //4.是否新增粉丝团
        DataStream<String> isNewFanTeamStream = isNewFanTeam(fanRelationStream);
        KafkaUtil.producerKafka(isNewFanTeamStream,"dwd_user_new_fans_team_detail");
        
        return isNewFanStream;
    }

    private static DataStream<String> isNewFanTeam(DataStream<String> stream) {
        KeyedStream<String, String> is_not_fan = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                return JSON.parseObject(s).getString("is_not_fan");
            }
        });
        SingleOutputStreamOperator<String> isNotFanProcess = is_not_fan.process(new IsNotFanFunction());

        return isNotFanProcess;
    }

    private static DataStream<String> isNewFan(DataStream<String> stream) {
        KeyedStream<String, String> isNewFanStream = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                return JSON.parseObject(s).getString("is_new");
            }
        });
        SingleOutputStreamOperator<String> isNewFanProcess = isNewFanStream.process(new IsNewFanFunction());

        return isNewFanProcess;
    }

    private static DataStream<String> isNewComment(DataStream<String> stream) {
        KeyedStream<String, String> isNewCommentStream = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String user_id = jsonObject.getString("user_id");
                String anchor_id = jsonObject.getString("anchor_id");
                String live_session_id = jsonObject.getString("live_session_id");
                return user_id+"_"+anchor_id+"_"+live_session_id;
            }
        });
        //状态编程，对is_new_comment进行校验恢复
        SingleOutputStreamOperator<String> isNewCommentProcess = isNewCommentStream.process(new IsNewCommentFunction());

        return isNewCommentProcess;
    }

    private static DataStream<String> isNewUserOrder(DataStream<String> stream) {
        KeyedStream<String, String> isFirstOrderStream = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                /*
                {
                "live_room_id": 1001,
                "anchor_id": 888,
                "sku_id": "sku_1006",
                "order_time": "2025-08-26 10:30:31",
                "event_id": "evt_187185",
                "user_id": "usr_30161",
                "order_cnt": 3,
                "is_first_order": 0,
                "order_amount": 5479.23,
                "live_session_id": "live_20250826_888_001",
                "sku_name": "海尔双开门冰箱",
                "category": "家用电器",
                "order_id": "ord_653141",
                "status": "finished"
                }
                 */
                return JSON.parseObject(s).getString("is_first_order");
            }
        });

        SingleOutputStreamOperator<String> isNewUserOrderProcess = isFirstOrderStream.process(new IsNewUserOrderFunction());

        return isNewUserOrderProcess;
    }
}
